package com.oracle.violet.first.report

import com.oracle.violet.first.config.ConfigHandler
import com.oracle.violet.first.until.Jpools
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import scala.util.control.Breaks._

object DataMark {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("数据标签化")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)

    //读取
    val dataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)
    dataFrame.mapPartitions(patitions=> {
      val jedis = Jpools.getJedis
      val data = patitions.map(line => {
        val adspacetype = line.getAs[Int]("adspacetype")
        val appid = line.getAs[String]("appid")
        var appname = line.getAs[String]("appname")
        var channelid = line.getAs[String]("channelid")
        var client = line.getAs[Int]("client")
        var networkmannername = line.getAs[String]("networkmannername")
        var ispname = line.getAs[String]("ispname")
        var keywords = line.getAs[String]("keywords")
        var provincename = line.getAs[String]("provincename")
        var cityname = line.getAs[String]("cityname")

        //获取用户标识字段
        var imie = line.getAs[String]("imei")
        val idfa = line.getAs[String]("idfa")
        val mac = line.getAs[String]("mac")
        val androidid = line.getAs[String]("androidid")
        val openudid = line.getAs[String]("openudid")
        val imiemd5 = line.getAs[String]("imeimd5")
        val idfamd5 = line.getAs[String]("idfamd5")
        val macmd5 = line.getAs[String]("macmd5")
        val androididmd5 = line.getAs[String]("androididmd5")
        val openudidmd5 = line.getAs[String]("openudidmd5")
        val imiesha1 = line.getAs[String]("imeisha1")
        val idfasha1 = line.getAs[String]("idfasha1")
        val macsha1 = line.getAs[String]("macsha1")
        val androididsha1 = line.getAs[String]("androididsha1")
        val openudidsha1 = line.getAs[String]("openudidsha1")

        var map = Map[String, Int]()

        //广告位类型
        if (adspacetype != 0 && adspacetype != null) {
          adspacetype match {
            case x if x < 10 => map += ("LC0" + x -> 1)
            case x if x > 9 => map += ("LC" + x -> 1)
          }
        }

        //App名称
        if (StringUtils.isEmpty(appname)) {
          if (StringUtils.isNotEmpty(appid)) {
            appname = jedis.hget("violet", appid)
            if (StringUtils.isEmpty(appname))
              map += ("app" + appname -> 1)
          }
        }

        //渠道
        if (StringUtils.isNotEmpty(channelid)) {
          map += ("CN".concat(channelid) -> 1)
        }

        //用户标识判断
        val uid = Array(imie, idfa, mac, androidid, openudid, imiemd5, idfamd5, macmd5, androididmd5,
          openudidmd5, imiesha1, idfasha1, macsha1, androididsha1, openudidsha1)
        var result = ""
        breakable(
          for (i <- uid) {

            val tureUid=line.getAs[String](i)
            if (StringUtils.isNotEmpty(tureUid)) {
              imie = tureUid
              break()
            }
          }
        )


        //设备
        //操作系统
        //设备操作系统 1 Android D00010001 2 IOS D00010002 3 WinPhone D00010003 _ 其他 D00010004

        var deviceDict = ""
        if (client == 1) {
          deviceDict = "D00010001"
        } else if (client == 2) {
          deviceDict = "D00010002"
        } else if (client == 3) {
          deviceDict = "D00010003"
        } else {
          deviceDict = "D00010004"
        }
        map += (deviceDict -> 1)

        //设备联网方式 WIFI D00020001 4G D00020002 3G D00020003 2G D00020004 _ D00020005
        if ("Wifi".equals(networkmannername)) {
          networkmannername = "D00020001"
        } else if ("4G".equals(networkmannername)) {
          networkmannername = "D00020002"
        } else if ("3G".equals(networkmannername)) {
          networkmannername = "D00020003"
        } else if ("2G".equals(networkmannername)) {
          networkmannername = "D00020004"
        } else {
          networkmannername = "D00020005"
        }
        map += (networkmannername -> 1)

        //设备运营商方式 移动 D00030001 联通 D00030002 电信 D00030003 _ D00030004
        if ("移动".equals(ispname)) {
          ispname = "D00030001"
        } else if ("联通".equals(ispname)) {
          ispname = "D00030002"
        } else if ("电信".equals(ispname)) {
          ispname = "D00030003"
        } else {
          ispname = "D00030004"
        }
        map += (ispname -> 1)


        //关键字
        if (StringUtils.isNotEmpty(keywords)) {
          val fields = keywords.split("\\|")
            .filter(word => {
              word.length >= 3 && word.length <= 8
            }).map(str => {
            map += ("K".concat(str.replace(":", "")) -> 1)
          })
        }

        //地域标签
        if (StringUtils.isNotEmpty("provincename")) {
          map += ("ZP" + provincename -> 1)
        }
        if (StringUtils.isNotEmpty(cityname)) {
          map += ("ZC" + cityname -> 1)
        }

        //上下文标签
        //如果esult不为空，获取数据
        (result, map)
      })
      jedis.close()
      data
    }).filter(x=>StringUtils.isNotEmpty(x._1))
      .reduceByKey((a,b)=>(a.zip(b).map(t=>(t._1._1,t._1._2+t._2._2))))
      .foreach(println)

      sc.stop()
  }
}
